package com.lcifn.middleware.rpc.transport;

import com.lcifn.middleware.rpc.transport.coder.ClientRequestEncoder;
import com.lcifn.middleware.rpc.transport.coder.ClientResponseDecoder;
import com.lcifn.middleware.rpc.transport.future.InFlightRequest;
import com.lcifn.middleware.rpc.transport.invocation.ClientResponseInvocation;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeoutException;

/**
 * 传输客户端netty实现
 * @author lichao
 * @date 2021/1/6 18:29
 */
public class NettyTransportClient implements TransportClient {
    private static final Logger logger = LoggerFactory.getLogger(NettyTransportClient.class);
    // 事件线程组
    private EventLoopGroup ioEventGroup = newEventLoopGroup();
    // channel集合
    private List<Channel> channels = new LinkedList<>();
    // 在途请求
    private InFlightRequest inFlightRequest;

    public NettyTransportClient() {
        inFlightRequest = new InFlightRequest();
    }

    /**
     * 创建传输服务通道
     *
     * @param socketAddress     socket地址
     * @param connectionTimeout 连接超时时间
     * @return
     */
    @Override
    public Transport createTransport(SocketAddress socketAddress, int connectionTimeout) throws TimeoutException, InterruptedException {
        return new NettyTransport(createChannel(socketAddress, connectionTimeout), inFlightRequest);
    }

    /**
     * 创建远程服务Channel
     * @param socketAddress
     * @param connectionTimeout
     * @return
     */
    private Channel createChannel(SocketAddress socketAddress, int connectionTimeout) throws InterruptedException, TimeoutException {
        // 创建Bootstrap
        ChannelHandler channelHandlerPipeline = newChannelPipelineHandler();
        Bootstrap bootstrap = newBootstrap(channelHandlerPipeline);

        // 连接服务端
        ChannelFuture channelFuture = bootstrap.connect(socketAddress);
        if (!channelFuture.await(connectionTimeout)) {
            throw new TimeoutException("connect to server " + socketAddress.toString() + " timeout for " + connectionTimeout + "ms!");
        }

        // 获取服务Channel
        Channel channel = channelFuture.channel();
        if (channel == null || !channel.isActive()) {
            throw new IllegalStateException("connect to server " + socketAddress.toString() + ", return channel is null or not active!");
        }

        // 添加到channel集合
        channels.add(channel);

        return channel;
    }

    /**
     * 创建Bootstrap
     * @param channelHandlerPipeline
     * @return
     */
    private Bootstrap newBootstrap(ChannelHandler channelHandlerPipeline) {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(getSocketChannelClass())
                .group(ioEventGroup)
                .handler(channelHandlerPipeline)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        return bootstrap;
    }

    /**
     * 创建事件线程组
     * @return
     */
    private EventLoopGroup newEventLoopGroup() {
        if (Epoll.isAvailable()) {
            return new EpollEventLoopGroup();
        }
        return new NioEventLoopGroup();
    }

    /**
     * 获取SocketChannel
     * @return
     */
    private Class<? extends Channel> getSocketChannelClass() {
        return Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class;
    }

    /**
     * 获取ChannelHandler
     * @return
     */
    private ChannelHandler newChannelPipelineHandler() {
        return new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline()
                        .addLast(new ClientRequestEncoder())
                        .addLast(new ClientResponseDecoder())
                        .addLast(new ClientResponseInvocation(inFlightRequest));
            }
        };
    }

    @Override
    public void close() throws IOException {
        for (Channel channel : channels) {
            channel.close();
        }

        if (ioEventGroup != null) {
            ioEventGroup.shutdownGracefully();
        }
    }
}
